package com.angel.bo.message.component.rocketmq;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.xiaoleilu.hutool.util.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;

/**
 * @author zhangwei
 * @date 2018/9/29 18:36
 * @desc 阿里rocket mq 工具类
 */
@Component
@Slf4j
public class ALiMqComponent {

    @Autowired
    private Producer producer;

    @Autowired
    private RedisTemplate redisTemplate;

    @Value("${ali.mq.topic}")
    private String topic;
    /**
     * 消息处理异常后的最大重试次数
     */
    private final int MAX_TRY_TIME = 3;

    /**
     * 发送消息
     *
     * @param tag
     * @param msg
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public SendResult send(String tag, Object msg) {
        try {
            return producer.send(aliMessage(tag, msg));
        } catch (Exception e) {
            saveExceptionInfo(tag, msg, e);
            return null;
        }
    }

    /**
     * 发送延时消息
     *
     * @param tag    tag
     * @param msg    消息体
     * @param second 延时时间,单位秒,时间最长不能超过40天
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public SendResult send(String tag, Object msg, long second) {
        try {
            return producer.send(aliMessage(tag, msg, second));
        } catch (Exception e) {
            saveExceptionInfo(tag, msg, e);
            return null;
        }
    }

    /**
     * 构建阿里mq消息体
     *
     * @param tag 标签
     * @param msg 消息体(消息体可以为任何对象,不限于String或int等类型)
     * @return
     * @throws Exception
     */
    private Message aliMessage(String tag, Object msg) {
        return new Message(topic, tag, ObjectUtil.serialize(msg));
    }

    /**
     * 构建阿里mq消息体
     *
     * @param tag    标签
     * @param msg    消息体(消息体可以为任何对象,不限于String或int等类型)
     * @param second 延时时间,单位:秒
     * @return
     * @throws Exception
     */
    private Message aliMessage(String tag, Object msg, long second) {
        Message message = new Message(topic, tag, ObjectUtil.serialize(msg));
        message.setStartDeliverTime(System.currentTimeMillis() + (second * 1000));
        return message;
    }

    /**
     * 处理mq集群订阅模式下的消费者异常信息
     * 注意:此方法不适用于广播订阅模式,因为广播模式下阿里mq不会进行消息重发
     * 广播模式应直接将错误信息存入数据库,参考方法:saveError(Message message, Exception e)
     *
     * @param e
     */
    @Transactional(rollbackFor = Exception.class)
    @SuppressWarnings("unchecked")
    public Action mqConsumerError(Message message, Exception e) {
        String incrKey = "MqError:" + message.getMsgID();
        Long num = (Long) redisTemplate.opsForValue().get(incrKey);
        if (num == null || num < MAX_TRY_TIME) {
            num = num == null ? 0 : num;
            redisTemplate.opsForValue().set(incrKey, num + 1);
            redisTemplate.expire(incrKey, 3600 * 5, TimeUnit.SECONDS);
            return Action.ReconsumeLater;
        } else {
            saveExceptionInfo(message, e);
            return Action.CommitMessage;
        }
    }

    /**
     * 保存异常信息
     *
     * @param message
     * @param e
     */
    private void saveExceptionInfo(Message message, Exception e) {
        log.error("ALiMq.send error--> tag:{};message:{}", message.getTag(), message.getBody());
        // TODO: 2018/10/17 这里自定义异常处理,如保存消息详情和异常信息
    }

    /**
     * 保存异常信息
     *
     * @param tag
     * @param msg
     * @param e
     */
    private void saveExceptionInfo(String tag, Object msg, Exception e) {
        log.error("ALiMq.send error--> tag:{};message:{}", tag, msg);
        // TODO: 2018/10/17 这里自定义异常处理,如保存消息详情和异常信息
    }
}